package server

import (
	"codeforces/src/config"
	"codeforces/src/task"
	"log"
	"os"
	"os/signal"

	"github.com/Shopify/sarama"
)

type Server struct {
}

func (server *Server) Run() {
	sigIntChan := make(chan os.Signal, 1)
	signal.Notify(sigIntChan, os.Interrupt)

	fil, _ := os.OpenFile("./log/log.log", os.O_CREATE|os.O_APPEND|os.O_RDWR, os.ModePerm)
	logger := log.New(fil, "", log.Ldate|log.Ltime)

	var conf = config.GetConfig()
	consumer, err := sarama.NewConsumer(conf.BrokerList, sarama.NewConfig())
	if err != nil {
		panic(err)
	}
	partitionPartition, err := consumer.ConsumePartition("cfReq", int32(conf.Partition), sarama.OffsetNewest)
	if err != nil {
		panic(err)
	}

	producer, err := sarama.NewAsyncProducer(conf.BrokerList, sarama.NewConfig())
	if err != nil {
		panic(err)
	}
	var (
		reqChanFetch      = make(chan task.ITask, 100)
		fetchChanParse    = make(chan task.ITask, 100)
		parseChanSendResp = make(chan task.ITask, 100)
	)
	interrupt := false
	done := make(chan struct{}, 1)
	checkDone := func() {
		if interrupt && len(reqChanFetch) == 0 && len(fetchChanParse) == 0 && len(parseChanSendResp) == 0 {
			done <- struct{}{}
		}
	}
	go handleReq(partitionPartition, reqChanFetch, logger)
	go fetch(reqChanFetch, fetchChanParse)
	go parse(fetchChanParse, parseChanSendResp)
	go sendResp(producer, parseChanSendResp, checkDone, logger)
	<-sigIntChan
	interrupt = true
	partitionPartition.Close()
	consumer.Close()
	<-done
	producer.Close()
	// pro

}

func CreateServer() *Server {
	return new(Server)
}
